package com.example.service;

import com.example.constant.MQConstant;
import com.example.po.SysMQMessage;
import com.example.startup.SysMQStartUpService;
import com.example.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author chentao
 * @version 1.0
 * @description: 用于主从架构MQ发送消息
 * @date 2025/3/1 上午12:36
 */
@Service
public class SysMQMessageASyncSendService implements InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private SysMQMessageService sysMQMessageService;

    //每次抓取总数
    @Value("${mq.message.max-size:10}")
    private int maxSize;

    //消息处理线程池核心数
    @Value("${mq.message.core-size:12}")
    private int corePoolSize;

    //消息处理间隔
    @Value("${mq.message.sec:5}")
    private int sec;

    //消息重试最大次数
    @Value("${mq.message.retry.count:10}")
    private int maxRetryCount;

    //重试间隔 60秒
    @Value("${mq.message.retry.period:60}")
    private int maxRetryPeriod;

    @Resource
    private SysServerService sysServerService;

    @Resource
    private SysMQStartUpService sysMQStartUpService;

    private ExecutorService executorService;

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    /****
     * 启动主节点发送MQ消息
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        logger.warn("server节点: " + sysServerService.getCurrentServerKey() + " 启动中 ...");
        this.executorService = Executors.newCachedThreadPool();
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(this.corePoolSize);
        this.scheduledThreadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    List<SysMQMessage> sysMQMessages = sysMQMessageService.getSysMQMessage(sysServerService.getCurrentServerKey(), maxSize, maxRetryCount);
                    if (!sysMQMessages.isEmpty()) {
                        CountDownLatch countDownLatch = new CountDownLatch(sysMQMessages.size());
                        for (SysMQMessage sysMQMessage : sysMQMessages) {
                            executorService.execute(new MQMessageSendThread(sysMQMessage, countDownLatch));
                        }
                        countDownLatch.await();
                    }
                } catch (Exception e) {
                    logger.warn("批量发送MQ消息失败: " + e.getMessage());
                }
            }
        }, 0, this.sec, TimeUnit.SECONDS);
    }

    /****
     * 关闭线程回收资源
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        this.scheduledThreadPoolExecutor.shutdownNow();
        this.executorService.shutdownNow();
    }

    /****
     * MQ消息推送
     */
    class MQMessageSendThread implements Runnable {

        private SysMQMessage sysMQMessage;

        private CountDownLatch countDownLatch;

        public MQMessageSendThread(SysMQMessage sysMQMessage, CountDownLatch countDownLatch) {
            this.sysMQMessage = sysMQMessage;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            int retryCount = sysMQMessage.getRetryCount();
            try {
                sysMQMessage.setStatus(MQConstant.SUCCESS);
                sysMQStartUpService.send(sysMQMessage.getType(), sysMQMessage.getName(), sysMQMessage.getContent());
            } catch (Exception e) {
                int nextRetryCount = retryCount + 1;
                Date nextRetryTime = TimeUtils.getTime(maxRetryPeriod * (nextRetryCount));
                sysMQMessage.setRetryTime(nextRetryTime);
                if (nextRetryCount == maxRetryCount) {
                    sysMQMessage.setStatus(MQConstant.ERROR);
                }else {
                    sysMQMessage.setStatus(MQConstant.INIT);
                }
                sysMQMessage.setErrorMessage(e.getMessage());
                logger.warn("发送MQ消息: {} 失败: {}", sysMQMessage, e.getMessage());
            } finally {
                sysMQMessage.setRetryCount(retryCount + 1);
                sysMQMessage.recalculateFields();
                sysMQMessageService.saveOrUpdate(sysMQMessage);
                countDownLatch.countDown();
            }
        }
    }
}
